package day07;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * @Description: 写入redis
 * @Author: ZYX
 * @Date: 2022/2/18 10:58
 * @Version: 1.0
 */
public class Demo03 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Tuple2<String, Integer>> stream = env.fromElements(
                Tuple2.of("key", 1),
                Tuple2.of("key", 2)
        );
        FlinkJedisPoolConfig redisConf = new FlinkJedisPoolConfig.Builder().setHost("20.18.27.177")
                .setPassword("123456").build();

        stream.addSink(new RedisSink<Tuple2<String,Integer>>(redisConf,new MyRedisMapper()));

        env.execute();
    }

    public static class MyRedisMapper implements RedisMapper<Tuple2<String,Integer>>{
        @Override
        public String getKeyFromData(Tuple2<String, Integer> tp2) {
            return tp2.f0;
        }

        @Override
        public String getValueFromData(Tuple2<String, Integer> tp2) {
            return tp2.f1.toString();
        }

        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET,"tuple:flink:redis:test");
        }
    }

}
